const { Observable, Subject, of, throwError, EMPTY, merge } = require('rxjs')
const { mergeMap,delay, map, tap, share, scan, takeUntil, catchError, ignoreElements } = require("rxjs/operators");
const { toPeer } = require("../dist/index")
const { Server } = require("ws");
const { xtp } = require("../src/pb")

// let arr = [1,[2,3],[4,[5,6]],7]
// function flat1(acc,task){
//     if (!task.length) return acc 
//     const [first, ...rest] = task 
//     if(Array.isArray(first)) return flat2(acc , [...first, ...rest])
//     return flat2([...acc, first], rest)
// }
// function flat2(acc,task){
//     if (!task.length) return acc.reverse()
//     const last = task[task.length-1]
//     const front = task.slice(0,task.length-1)
//     if(Array.isArray(last)) return flat3(acc , front.concat(last))
//     return flat3(acc.concat(last), front)
// }

//服务方法路由
const router = {
    "acc": onAcc,
    "delayAcc": onDelayAcc,
}

//创建ws服务器
webSocketServer({
    port: 8000,
    handleProtocols: (protocols) => {
        console.log(protocols)
        return protocols
    }
}).pipe(
    tap(ws=>console.log("on ws")),
    map(toConnection),
    map(toPeer),
    mergeMap(onPeer),
    catchError(err=>EMPTY),
).subscribe()

//创建根处理者
function onPeer(peer) {
    console.log("onPeer")
    //声明根处理者信息
    const info = {
        data: Buffer.from("Service Info"),
        channelSize: 10,
    }
    const rootHandler = peer.createHandler(info)
    const declare = of(rootHandler).pipe(
        tap(it=>peer.declareRootHandlerInfo(it)),
        ignoreElements(),
    )
    //处理输入，这里是简单的路由逻辑
    const input = rootHandler.onData.pipe(
        mergeMap(data=>{
            console.log("input data:", data, data.toString())
            const fn = router[data.toString()]
            if (!fn)
                return EMPTY
            return fn(peer, rootHandler)
        }),
    )
    return merge(
        input,
        declare,
    )
}


//累加收到的请求个数，响应json字符串，形如{"time":"2021-03-01 10:31:59","acc":13}
function onAcc(peer, rootHandler) {
    console.log("onAcc")
    //当收到"acc"请求时，产出（声明）"accHandler"处理者信息
    const info = {
        data: Buffer.from("accHandler"),
        channelSize: 1000,
    }
    const handler = peer.createHandler(info)
    const declare = of(handler).pipe(
        tap(it=>rootHandler.yieldHandler(it)),
        ignoreElements(),
    )
    //把接收的data产出（变换）为json字符串并输出
    const output = handler.onData.pipe(
        // tap(data=>console.log("data:",data)),
        scan((acc, val) => acc + 1, 0),
        map(acc => {
            // const buf = Buffer.alloc(4)
            // buf.writeUInt32BE(acc)
            const json = JSON.stringify({ time: new Date().toLocaleString(), acc })
            // console.log("json:",json)
            const buf = Buffer.from(json)
            return buf
        }),
        tap(it=>handler.yieldedDataSender.next(it)),
        ignoreElements()
    )
    return merge(output, declare)
}

function onDelayAcc(peer, rootHandler) {
    console.log("onDelayAcc")
    //当收到"delayAcc"请求时，产出（声明）"delayAccHandler"处理者信息
    const info = {
        data: Buffer.from("delayAccHandler"),
        channelSize: 1,
    }
    const handler = peer.createHandler(info)
    const declare = of(handler).pipe(
        tap(it=>rootHandler.yieldHandler(it)),
        ignoreElements(),
    )
    //把接收的data产出（变换）为json字符串并输出
    const output = handler.onData.pipe(
        // tap(data=>console.log("data:",data)),
        delay(1000),
        scan((acc, val) => acc + 1, 0),
        map(acc => {
            // const buf = Buffer.alloc(4)
            // buf.writeUInt32BE(acc)
            const json = JSON.stringify({ time: new Date().toLocaleString(), acc })
            // console.log("json:",json)
            const buf = Buffer.from(json)
            return buf
        }),
        tap(it=>handler.yieldedDataSender.next(it)),
        ignoreElements()
    )
    return merge(output, declare)
}

function webSocketServer(options) {
    return new Observable(s => {
        const wss = new Server(options)
        wss.on('close', function close() {
            s.error("close")
        })
        wss.on('connection', function connection(ws) {
            s.next(ws)
        })
    })
}
function toConnection(ws) {
    const theEnd = new Observable(s => {
        ws.on('close', function (code, reason) {
            // s.error(reason)
            s.next(null)
            s.complete()
        })
    })
    const onFrame = new Observable(s => {
        // console.log('on sub buffers')
        const cb = (data) => {
            // console.log('received:', data);
            const frame = xtp.Frame.decode(data)
            s.next(frame)
        }
        ws.addListener("message", cb)
        return () => {
            console.log('on unsub buffers')
            ws.removeListener("message", cb)
        }
    }).pipe(
        // finalize(()=>console.log('buffers finalize')),
        takeUntil(theEnd),
        share(),
    )
    const frameSender = new Subject()
    frameSender.pipe(
        takeUntil(theEnd),
    ).subscribe({
        next:frame => {
            const buf = xtp.Frame.encode(frame).finish()
            ws.send(buf)
            // console.log("send:", buf)
        },
        error:err => {
            ws.close(4000, err.toString())
            // console.log("close ws:", err)
        },
        complete:() => {
            ws.close()
        }
    })
    return {
        onFrame,
        frameSender,
    }
}
